#pragma once

#include "errors.h"
#include <cstring>
#include <fcntl.h>
#include <sys/mman.h>
#include <unistd.h>
#include <vector>
#include <algorithm>
#include <cassert>
#include <cmath>
#include <cstdlib>
#include <fstream>
#include <ios>
#include <memory>
#include "system/sm_meta.h"
class MergeSort {
private:
    int max_mem_size;
    int rcd_size;

    int len_ = 0;

    int index = 0;
    char *data = nullptr;
    bool isFull = true;

public:
    MergeSort(int rcd_size, void *cols = nullptr)  :  rcd_size(rcd_size), cols_(cols) {}

    MergeSort() = default;

    int getLen(){
        return len_;
    }

    void beginSort(const char *record) {
        max_mem_size = (MAX_MEMORY - MAX_MEMORY % rcd_size);
        if (isFull) open_new_temp_file();
        memcpy(data + index * rcd_size, record, rcd_size);
        index++;
        len_++;
        if (index == max_mem_size / rcd_size) isFull = true;
    }

    void endSort() {
        if (data != nullptr) sort_and_write_data();
        open_and_buffer_files();

        size_t height = std::ceil(std::log2(filenames_.size()));
        std::vector<int> winners;

        initialize_heap_and_winners(winners, height);
        build_loser_tree(winners, height);
    }

    void nextRcd(char *record) {
        memcpy(record, record_list[heap[0]].get(), rcd_size);
        adjust();
        len_--;
    }

private:
    
    void adjust() {
        int file_index = heap[0];
        read_next_record(file_index);
        size_t height = std::ceil(std::log2(filenames_.size()));
        int cur = file_index + (1 << height);
        int winner = file_index;

        if (is_file_eof(file_index)) {
            close_and_cleanup_file(file_index, cur);
            winner = -1;
        }

        adjust_heap(winner, cur);
    }

    std::vector<std::unique_ptr<char[]>> record_list;
    void read_next_record(int file_index) {
        opened_files[file_index].read(record_list[file_index].get(), rcd_size);
    }

    bool is_file_eof(int file_index) {
        return opened_files[file_index].eof();
    }

    std::vector<int> heap;
    void close_and_cleanup_file(int file_index, int cur) {
        heap[cur] = -1;
        opened_files[file_index].close();
        unlink(filenames_[file_index].c_str());
    }

    static int cmp_(const void *a, const void *b, void *arg) {
        auto *col = static_cast<const ColMeta *>(arg);
        auto lvalue = convertValue(static_cast<char *>(const_cast<void *>(a)), *col);
        auto rvalue = convertValue(static_cast<char *>(const_cast<void *>(b)), *col);
        return (lvalue < rvalue) ? -1 : (lvalue > rvalue) ? 1 : 0;
    }

    void *cols_;
    void adjust_heap(int winner, int cur) {
        while (cur != 1) {
            int parent = cur >> 1;
            if (winner != -1 && (heap[parent] == -1 || cmp_(record_list[winner].get(), record_list[heap[parent]].get(), cols_) <= 0)) {
                cur = parent;
            } else {
                std::swap(heap[parent], winner);
                cur = parent;
            }
        }
        heap[0] = winner;
    }

    void sort_and_write_data() {
        qsort_r(data, index, rcd_size, cmp_, cols_);
        munmap(data, max_mem_size);
        int fd = open(filenames_.back().c_str(), O_RDWR);
        if (fd == -1) throw UnixError();
        ftruncate(fd, index * rcd_size);
        close(fd);
    }

    void open_new_temp_file() {
        if (data != nullptr) sort_and_write_data();
        char filename[] = "tempXXXXXX";
        int fd = mkstemp(filename);
        if (fd == -1) throw UnixError();
        filenames_.emplace_back(filename);
        ftruncate(fd, max_mem_size);
        data = static_cast<char *>(mmap(nullptr, max_mem_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0));
        if (data == MAP_FAILED) throw UnixError();
        isFull = false;
        index = 0;
        close(fd);
    }

    void build_loser_tree(std::vector<int>& winners, size_t height) {
        for (int i = (1 << height) - 1; i >= 1; i--) {
            int left = i << 1;
            int right = i << 1 ^ 1;
            if (winners[left] != -1 && (winners[right] == -1 || cmp_(record_list[winners[left]].get(),
                                                                     record_list[winners[right]].get(), cols_) <= 0)) {
                winners[i] = winners[left];
                heap[i] = winners[right];
            } else {
                winners[i] = winners[right];
                heap[i] = winners[left];
            }
        }
        heap[0] = winners[1];
    }

    std::vector<std::unique_ptr<char[]>> buffer_list;
    std::vector<std::ifstream> opened_files;
    void open_and_buffer_files() {
        int buffer_size = (MAX_MEMORY / filenames_.size()) & ~(rcd_size - 1);

        for (const auto &filename : filenames_) {
            std::ifstream file(filename, std::ios::binary);
            if (file.fail()) throw UnixError();

            auto buffer = std::make_unique<char[]>(buffer_size);
            file.rdbuf()->pubsetbuf(buffer.get(), buffer_size);
            buffer_list.push_back(std::move(buffer));

            auto record = std::make_unique<char[]>(rcd_size);
            file.read(record.get(), rcd_size);
            record_list.push_back(std::move(record));

            opened_files.push_back(std::move(file));
        }
    }

    std::vector<std::string> filenames_;
    void initialize_heap_and_winners(std::vector<int>& winners, size_t height) {
        size_t num_nodes = 1 << (height + 1);

        heap.assign(num_nodes, -1);
        winners.assign(num_nodes, -1);

        for (size_t i = 0; i < filenames_.size(); ++i) {
            size_t pos = (1 << height) + i;
            heap[pos] = winners[pos] = i;
        }
    }

public:
    MergeSort(MergeSort&& sorter) noexcept
    :  rcd_size(sorter.rcd_size), len_(sorter.len_),
        index(sorter.index), data(sorter.data),
        isFull(sorter.isFull), record_list(std::move(sorter.record_list)), heap(std::move(sorter.heap)),
        cols_(sorter.cols_), buffer_list(std::move(sorter.buffer_list)), opened_files(std::move(sorter.opened_files)), filenames_(std::move(sorter.filenames_)) {}

    MergeSort& operator=(MergeSort&& sorter) noexcept {
        rcd_size = sorter.rcd_size;
        len_ = sorter.len_;
        index = sorter.index;
        data = sorter.data;
        isFull = sorter.isFull;
        record_list = std::move(sorter.record_list);
        heap = std::move(sorter.heap);
        cols_ = sorter.cols_;
        buffer_list = std::move(sorter.buffer_list);
        opened_files = std::move(sorter.opened_files);
        filenames_ = std::move(sorter.filenames_);
        return *this;
    }

    ~MergeSort() {
        for (size_t i = 0; i < opened_files.size(); ++i) {
            if (opened_files[i].is_open()) {
                opened_files[i].close();
                unlink(filenames_[i].c_str());
            }
        }
    }
};
